package com.atguigu.flink.timeAndwindow;

import com.atguigu.flink.pojo.OrderDetailEvent;
import com.atguigu.flink.pojo.OrderEvent;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.time.Duration;

/**
 * Created by 黄凯 on 2023/6/18 0018 20:44
 *
 * @author 黄凯
 * 永远相信美好的事情总会发生.
 * <p>
 * 窗口联结
 * *
 * * 能进入到同一个窗口的数据才有资格进行join ，如果来自两条流的数据的key一样，就可以join到一起。
 */
public class Flink11_WindowJoin {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //orderEvent
        //order-1,1000
        SingleOutputStreamOperator<OrderEvent> orderDs = env.socketTextStream("127.0.0.1", 8888)
                .map(
                        line -> {
                            String[] fields = line.split(",");
                            return new OrderEvent(fields[0].trim(), Long.valueOf(fields[1].trim()));
                        }

                ).assignTimestampsAndWatermarks(
                        WatermarkStrategy.<OrderEvent>forBoundedOutOfOrderness(Duration.ZERO)
                                .withTimestampAssigner(
                                        (event, ts) -> event.getTs()
                                )
                );

        orderDs.print("orderDs-input");

        //OrderDetailEvent
        //detail-1,order-1,Apple,3000
        SingleOutputStreamOperator<OrderDetailEvent> detailDs = env.socketTextStream("127.0.0.1", 9999)
                .map(
                        line -> {
                            String[] fields = line.split(",");
                            return new OrderDetailEvent(fields[0].trim(), fields[1].trim(), fields[2].trim(), Long.valueOf(fields[3].trim()));
                        }

                ).assignTimestampsAndWatermarks(
                        WatermarkStrategy.<OrderDetailEvent>forBoundedOutOfOrderness(Duration.ZERO)
                                .withTimestampAssigner(
                                        (event, ts) -> event.getTs()
                                )
                );

        detailDs.print("detailDs-input");

        // order流 与 detail流的join
        orderDs.join(
                detailDs
        ).where(
                OrderEvent::getOrderId
        ).equalTo(
                OrderDetailEvent::getOrderId
        ).window(
                TumblingEventTimeWindows.of(Time.seconds(10))
        ).apply(
                new JoinFunction<OrderEvent, OrderDetailEvent, String>() {

                    /**
                     * 处理join成功的数据
                     * @param first The element from first input.
                     * @param second The element from second input.
                     * @return
                     * @throws Exception
                     */
                    @Override
                    public String join(OrderEvent first, OrderDetailEvent second) throws Exception {
                        return first + " -- " + second;
                    }
                }
        ).print();

        env.execute();

    }

}
